package com.dmai.capacity.platform.gateway.listener;

import com.alibaba.fastjson.JSON;
import com.dmai.capacity.platform.common.lock.DistributedLock;
import com.dmai.capacity.platform.common.lock.LockAdapter;
import com.dmai.capacity.platform.common.model.CapabilityPackageDeductionRule;
import com.dmai.capacity.platform.common.model.GatewayLog;
import com.dmai.capacity.platform.common.model.OauthUserPackage;
import com.dmai.capacity.platform.common.utils.DateUtils;
import com.dmai.capacity.platform.common.utils.SpringUtil;
import com.dmai.capacity.platform.gateway.config.GatewayLogSinkConfig;
import com.dmai.capacity.platform.gateway.constants.RedisKeyConstant;
import com.dmai.capacity.platform.oauth.common.constants.ResourcePackageTypeEnum;
import com.dmai.capacity.platform.oauth.common.mapper.CapabilityPackageDeductionRuleDao;
import com.dmai.capacity.platform.oauth.common.service.IGatewayLogService;
import com.dmai.capacity.platform.oauth.common.service.ICapabilityPackageService;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.redisson.Redisson;
import org.redisson.api.RAtomicLong;
import org.redisson.api.RBucket;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.AutoConfigureBefore;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.config.BindingServiceConfiguration;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;


/**
 * 请求日志实现类-kafka
 */
@Slf4j
@Service
@AllArgsConstructor
@ConditionalOnClass(BindingServiceConfiguration.class)
@AutoConfigureBefore({BindingServiceConfiguration.class})
@EnableBinding(GatewayLogSinkConfig.Sink.class)
public class KafkaCapabilityCallbackServiceListener {

    @Autowired
    private IGatewayLogService gatewayLogService;

    @Autowired
    private ICapabilityPackageService resourcePackageService;

    @Resource
    CapabilityPackageDeductionRuleDao deductionRuleDao;

    @Autowired
    private Redisson redisson;

    @Autowired
    private DistributedLock locker;

    @StreamListener(target = GatewayLogSinkConfig.Sink.INPUT)
    public void handle(Message<String> message) throws Exception {
        log.info("KafkaAuditServiceListener received message : {}", message.getPayload());

        Acknowledgment acknowledgment = message.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class);

        if (acknowledgment != null) {
            CalligraphyHandleResult result = JSON.parseObject(message.getPayload(), CalligraphyHandleResult.class);
            int count = gatewayLogService.handleResponse(result.getSessionId(), result.getCode(), message.getPayload());
            // 更新资源包套餐
            GatewayLog log = gatewayLogService.get(result.getSessionId());
            if (log.getPackageId() > 0) {
                if (count > 0 && result.getCode().equals(0)) {
                    // packageId>0 表示本次请求走资源包，需要更新资源包使用情况
                    resourcePackageService.deduction(log.getCapabilityId(), log.getPackageId());
                } else {
                    try (LockAdapter lockAdapter = locker.lock(RedisKeyConstant.LOCK_USER_CAPABILITY_PERMISSION_PREFIX + ":" + log.getUserid() + ":" + log.getCapabilityId());) {
                        RBucket<List<OauthUserPackage>> cache = redisson.getBucket(RedisKeyConstant.CACHE_USER_PACKAGE_PREFIX + ":" + log.getUserid() + ":" + log.getCapabilityId());

                        List<OauthUserPackage> orig = cache.get();
                        orig.stream().filter(t -> t.getPackageId().equals(log.getPackageId())).forEach(t -> {
                            int _deduction = 1;
                            if (t.getType().equals(ResourcePackageTypeEnum.SHARE.getCode())) {
                                // 共享资源包
                                CapabilityPackageDeductionRule rule = deductionRuleDao.find(log.getCapabilityId());
                                _deduction = rule.getCount();
                            }

                            t.setQuotaAvailable(t.getQuotaAvailable() + _deduction);
                            t.setEffective(1);
                        });
                        cache.set(orig, cache.remainTimeToLive() / 1000, TimeUnit.SECONDS);
                    }
                }
                RAtomicLong atomicLong = redisson.getAtomicLong(RedisKeyConstant.CACHE_USER_REQUEST_COUNT_PREFIX + ":" + log.getUserid() + ":" + log.getCapabilityId());
                if (atomicLong.isExists()) {
                    atomicLong.decrementAndGet();
                }
            }
            //手动提交数据
            acknowledgment.acknowledge();
        }
    }
}
